package rabbitMQ

import (
	"fmt"
	"log"
	"time"

	"github.com/streadway/amqp"
	"github.com/zeromicro/go-zero/core/logx"
)

// Rabbit RabbitMQ结构体
type Rabbit struct {
	conn          *amqp.Connection
	channel       *amqp.Channel
	notifyConfirm chan amqp.Confirmation // 确认发送到mq的channel
	QueueName     string                 // 队列名称
	Exchange      string                 // 交换机名称
	Key           string                 // bind Key 名称
	MqUrl         string                 // 连接信息
}

// NewRabbitMQ 创建Rabbit结构体实例
func NewRabbitMQ(queueName, exchange, key, rmqURL string) *Rabbit {
	return &Rabbit{
		QueueName: queueName,
		Exchange:  exchange,
		Key:       key,
		MqUrl:     rmqURL,
	}
}

// Destroy 断开channel和connection
func (r *Rabbit) Destroy() error {
	err := r.channel.Close()
	err = r.conn.Close()
	return err
}

// 错误处理函数
func (r *Rabbit) failOnErr(err error, msg string) {
	if err != nil {
		log.Fatal(msg, err)
	}
}

// NewRabbitMQTopics 创建Topics模式下RabbitMQ实例
func NewRabbitMQTopics(queueName, exchangeName, routingKey, rmqURL string) *Rabbit {
	rabbitMQ := NewRabbitMQ(queueName, exchangeName, routingKey, rmqURL) // 创建RabbitMQ实例
	var err error
	rabbitMQ.conn, err = amqp.Dial(rabbitMQ.MqUrl) // 获取connection
	if err != nil {
		logx.Info("failed to connect rabbitmq!", err)
	}
	// rabbitMQ.failOnErr(err, "failed to connect rabbitmq!")
	rabbitMQ.channel, err = rabbitMQ.conn.Channel() // 获取channel
	if err != nil {
		logx.Info("failed to open a channel", err)
	}
	// rabbitMQ.failOnErr(err, "failed to open a channel")
	err = rabbitMQ.channel.Confirm(false)
	if err != nil {
		logx.Info("this.Channel.Confirm ", err)
	}
	rabbitMQ.notifyConfirm = rabbitMQ.channel.NotifyPublish(make(chan amqp.Confirmation, 1))
	go rabbitMQ.listenConfirm(queueName)
	return rabbitMQ
}

func (r *Rabbit) listenConfirm(queueName string) {
	for ret := range r.notifyConfirm {
		if ret.Ack {
			logx.Info("confirm: 消息发送成功", queueName)
		} else {
			logx.Info("confirm: 消息发送失败", queueName)
		}
	}
}

// PublishTopics Topics模式 生产者
func (r *Rabbit) PublishTopics(msg []byte) {
	// 1.尝试创建交换机
	err := r.channel.ExchangeDeclare(
		r.Exchange,         // 交换机名字
		amqp.ExchangeTopic, // 交换机类型，这里使用topic类型，即: Topics模式
		true,               // 是否持久化
		false,              // 是否自动删除
		false,              // true表示这个exchange不可以被client用来推送消息，仅用来进行exchange和exchange之间的绑定
		false,              // 是否阻塞处理
		nil,                // 额外的属性
	)
	logx.Info(err, "Failed to declare an exchange")
	// r.failOnErr(err, "Failed to declare an exchange")
	// 2.发送消息
	err = r.channel.Publish(
		r.Exchange,
		r.Key, // Topics模式这里要指定key
		false, // 如果为true，根据自身exchange类型和routekey规则无法找到符合条件的队列会把消息返还给发送者
		false, // 如果为true，当exchange发送消息到队列后发现队列上没有消费者，则会把消息返还给发送者
		amqp.Publishing{
			DeliveryMode: amqp.Persistent,
			ContentType:  "application/json",
			Body:         msg,
		},
	)
	if err != nil {
		log.Println(err)
	}
}

// PublishTopics Topics模式 生产者
func (r *Rabbit) PublishDirect(msg []byte, durable bool) {
	// 1.尝试创建交换机
	err := r.channel.ExchangeDeclare(
		r.Exchange,          // 交换机名字
		amqp.ExchangeDirect, // 交换机类型，这里使用Direct类型，即: Direct模式
		durable,             // 是否持久化
		false,               // 是否自动删除
		false,               // true表示这个exchange不可以被client用来推送消息，仅用来进行exchange和exchange之间的绑定
		false,               // 是否阻塞处理
		nil,                 // 额外的属性
	)
	logx.Info(err, "Failed to declare an exchange")
	// r.failOnErr(err, "Failed to declare an exchange")
	// 2.发送消息
	err = r.channel.Publish(
		r.Exchange,
		r.Key, // Topics模式这里要指定key
		false, // 如果为true，根据自身exchange类型和routekey规则无法找到符合条件的队列会把消息返还给发送者
		false, // 如果为true，当exchange发送消息到队列后发现队列上没有消费者，则会把消息返还给发送者
		amqp.Publishing{
			DeliveryMode: amqp.Persistent,
			ContentType:  "application/json",
			Body:         msg,
		},
	)
	if err != nil {
		log.Println(err)
	}else{
		logx.Info("发布成功")
	}
}

// PublishDeadTopics Topics模式 生产者
func (r *Rabbit) PublishDeadTopics(msg []byte, toExchange, toRouteKey string, ttl string) {
	// 1.试探性创建延迟交换机
	err := r.channel.ExchangeDeclare(
		r.Exchange,          // 交换机名字
		"x-delayed-message", // 交换机类型，这里使用topic类型，即: Topics模式
		true,                // 是否持久化
		false,               // 是否自动删除
		false,               // true表示这个exchange不可以被client用来推送消息，仅用来进行exchange和exchange之间的绑定
		false,               // 是否阻塞处理
		amqp.Table{
			"x-delayed-type": amqp.ExchangeDirect,
		}, // 额外的属性
	)
	logx.Info(err, "Failed to declare an exchange", r.Exchange)
	err = r.channel.ExchangeDeclare(
		toExchange,          // 交换机名字
		amqp.ExchangeDirect, // 交换机类型，这里使用topic类型，即: Topics模式
		true,                // 是否持久化
		false,               // 是否自动删除
		false,               // true表示这个exchange不可以被client用来推送消息，仅用来进行exchange和exchange之间的绑定
		false,               // 是否阻塞处理
		nil,                 // 额外的属性
	)
	// r.failOnErr(err, "Failed to declare an exchange")
	logx.Info(err, "Failed to declare an exchange", toExchange)
	// 2.试探性创建队列，这里注意队列名称不要写
	var args = make(map[string]interface{})
	args["x-dead-letter-exchange"] = toExchange    // 指定死信交换机
	args["x-dead-letter-routing-key"] = toRouteKey // 指定死信routing-key
	queue, err := r.channel.QueueDeclare(
		r.QueueName, // 随机生产队列名称
		true,        // 是否持久化
		false,       // 是否自动删除
		false,       // 是否具有排他性
		false,       // 是否阻塞处理
		args,        // 额外的属性
	)
	logx.Info(err, "Failed to declare a queue", r.QueueName)
	// 3.绑定队列到exchange中
	err = r.channel.QueueBind(
		queue.Name, // 队列名
		r.Key,      // 路由参数，如果匹配消息发送的时候指定的路由参数，消息就投递到当前队列（在Topics模式下，这里的key要指定）
		r.Exchange, // 交换机名字，需要跟消息发送端定义的交换器保持一致
		false,      // 是否阻塞处理
		nil,        // 额外的属性
	)
	_, err = r.channel.QueueDeclare(
		toExchange, // 随机生产队列名称
		true,       // 是否持久化
		false,      // 是否自动删除
		false,      // 是否具有排他性
		false,      // 是否阻塞处理
		nil,        // 额外的属性
	)
	err = r.channel.QueueBind(
		toExchange, // 队列名
		toRouteKey, // 路由参数，如果匹配消息发送的时候指定的路由参数，消息就投递到当前队列（在Topics模式下，这里的key要指定）
		toExchange, // 交换机名字，需要跟消息发送端定义的交换器保持一致
		false,      // 是否阻塞处理
		nil,        // 额外的属性
	)
	logx.Info(err, "Failed to declare a queue", toExchange)
	// r.failOnErr(err, "Failed to declare an exchange")
	// 2.发送消息
	err = r.channel.Publish(
		r.Exchange,
		r.Key, // Topics模式这里要指定key
		false, // 如果为true，根据自身exchange类型和routekey规则无法找到符合条件的队列会把消息返还给发送者
		false, // 如果为true，当exchange发送消息到队列后发现队列上没有消费者，则会把消息返还给发送者
		amqp.Publishing{
			// Expiration:   "100",
			DeliveryMode: amqp.Persistent,
			ContentType:  "application/json",
			Body:         msg,
			Headers: map[string]interface{}{
				"x-delay": ttl, // 消息从交换机过期时间,毫秒（x-dead-message插件提供）
			},
		},
	)
	if err != nil {
		log.Println(err)
	}
}

func (r *Rabbit) ConsumeTopics(f func(body string) (bool, error)) {
	defer func(f func(body string) (bool, error)) {
		if err := recover(); err != nil {
			time.Sleep(3 * time.Second)
			fmt.Println("sleep: 3 seconds", err)
		}
		r = NewRabbitMQTopics(r.QueueName, r.Exchange, r.Key, r.MqUrl)
		r.ConsumeTopics(f)
	}(f)

	err := r.channel.ExchangeDeclare(
		r.Exchange,         // 交换机名字
		amqp.ExchangeTopic, // 交换机类型，这里使用topic类型，即: Topics模式
		true,               // 是否持久化
		false,              // 是否自动删除
		false,              // true表示这个exchange不可以被client用来推送消息，仅用来进行exchange和exchange之间的绑定
		false,              // 是否阻塞处理
		nil,                // 额外的属性
	)
	logx.Info(err, "Failed to declare an exchange")
	// 2.试探性创建队列，这里注意队列名称不要写
	queue, err := r.channel.QueueDeclare(
		r.QueueName, // 随机生产队列名称
		true,        // 是否持久化
		false,       // 是否自动删除
		false,       // 是否具有排他性
		false,       // 是否阻塞处理
		nil,         // 额外的属性
	)
	logx.Info(err, "Failed to declare a queue")
	// 3.绑定队列到exchange中
	err = r.channel.QueueBind(
		queue.Name, // 队列名
		r.Key,      // 路由参数，如果匹配消息发送的时候指定的路由参数，消息就投递到当前队列（在Topics模式下，这里的key要指定）
		r.Exchange, // 交换机名字，需要跟消息发送端定义的交换器保持一致
		false,      // 是否阻塞处理
		nil,        // 额外的属性
	)
	defer r.channel.Close()
	closeChan := make(chan *amqp.Error, 1)
	notifyClose := r.channel.NotifyClose(closeChan)
	closeFlag := false
	// 4.消费消息
	msgs, err := r.channel.Consume(
		queue.Name, // 队列名称
		"",         // 用来区分多个消费者
		false,      // 是否自动应答
		false,      // 是否独有
		false,      // 设置为true，表示不能将同一个Connection中生产者发送的消息传递给这个Connection中的消费者
		false,      // 队列是否阻塞
		nil,        // 额外的属性
	)
	logx.Info(err, "Failed to Consume")
	// r.failOnErr(err, "Failed to Consume")
	// 5.启用协程处理消息
	for {
		select {
		case e := <-notifyClose:
			fmt.Println("chan error,e:%s", e.Error())
			close(closeChan)
			time.Sleep(5 * time.Second)
			r = NewRabbitMQTopics(r.QueueName, r.Exchange, r.Key, r.MqUrl)
			r.ConsumeTopics(f)
			closeFlag = true
		case delivery := <-msgs:
			flag, err := f(string(delivery.Body))
			if err != nil {
				// r.failOnErr(err, "Failed to Consume")
				logx.Info(err, "Failed to Consume")
				flag = false
			}
			if flag {
				// delivery.Reject(false)
				delivery.Ack(false)
				logx.Info("confirm: Topics消息消费成功", string(delivery.Body))
			} else {
				delivery.Ack(false)
				logx.Info("confirm: Topics消息消费失败", string(delivery.Body), err)
			}
		}
		if closeFlag {
			break
		}
	}
	// 5.启用协程处理消息
	// forever := make(chan bool) // 开个channel阻塞住，让开启的协程能一直跑着
	// go func() {
	// 	for delivery := range msgs {
	// 		// 消息逻辑处理，可以自行设计逻辑
	// 		// fmt.Println("Received a message:", string(delivery.Body))
	// 		flag, err := f(string(delivery.Body))
	// 		if err != nil {
	// 			// r.failOnErr(err, "Failed to Consume")
	// 			logx.Info(err, "Failed to Consume")
	// 			flag = false
	// 		}
	// 		if flag {
	// 			// delivery.Reject(false)
	// 			delivery.Ack(false)
	// 			logx.Info("confirm: 消息消费成功", string(delivery.Body))
	// 		} else {
	// 			delivery.Ack(false)
	// 			logx.Info("confirm: 消息消费失败", string(delivery.Body), err)
	// 		}
	// 		// delivery.Ack(false)
	// 	}
	// }()
	// fmt.Println(" [*] Waiting for messages.")
	// <-forever
}

func (r *Rabbit) ConsumeDirect(f func(body string) (bool, error), durable bool) {

	defer func(f func(body string) (bool, error), durable bool) {
		if err := recover(); err != nil {
			time.Sleep(3 * time.Second)
			fmt.Println("sleep: 3 seconds", err)
		}
		r = NewRabbitMQTopics(r.QueueName, r.Exchange, r.Key, r.MqUrl)
		r.ConsumeDirect(f, durable)
	}(f, durable)
	err := r.channel.ExchangeDeclare(
		r.Exchange,          // 交换机名字
		amqp.ExchangeDirect, // 交换机类型，这里使用Direct类型，即: Direct模式
		durable,             // 是否持久化
		false,               // 是否自动删除
		false,               // true表示这个exchange不可以被client用来推送消息，仅用来进行exchange和exchange之间的绑定
		false,               // 是否阻塞处理
		nil,                 // 额外的属性
	)
	logx.Info(err, "Failed to declare an exchange")
	// 2.试探性创建队列，这里注意队列名称不要写
	queue, err := r.channel.QueueDeclare(
		r.QueueName, // 随机生产队列名称
		true,        // 是否持久化
		false,       // 是否自动删除
		false,       // 是否具有排他性
		false,       // 是否阻塞处理
		nil,         // 额外的属性
	)
	logx.Info(err, "Failed to declare a queue")
	// 3.绑定队列到exchange中
	err = r.channel.QueueBind(
		queue.Name, // 队列名
		r.Key,      // 路由参数，如果匹配消息发送的时候指定的路由参数，消息就投递到当前队列（在Direct模式下，这里的key要指定）
		r.Exchange, // 交换机名字，需要跟消息发送端定义的交换器保持一致
		false,      // 是否阻塞处理
		nil,        // 额外的属性
	)

	defer r.channel.Close()
	closeChan := make(chan *amqp.Error, 1)
	notifyClose := r.channel.NotifyClose(closeChan)
	closeFlag := false
	// 4.消费消息
	msgs, err := r.channel.Consume(
		queue.Name, // 队列名称
		"",         // 用来区分多个消费者
		false,      // 是否自动应答
		false,      // 是否独有
		false,      // 设置为true，表示不能将同一个Connection中生产者发送的消息传递给这个Connection中的消费者
		false,      // 队列是否阻塞
		nil,        // 额外的属性
	)
	logx.Info(err, "Failed to Consume")
	// r.failOnErr(err, "Failed to Consume")
	// 5.启用协程处理消息
	for {
		select {
		case e := <-notifyClose:
			fmt.Println("chan error,e:%s", e.Error())
			close(closeChan)
			time.Sleep(5 * time.Second)
			r = NewRabbitMQTopics(r.QueueName, r.Exchange, r.Key, r.MqUrl)
			r.ConsumeDirect(f, durable)
			closeFlag = true
		case delivery := <-msgs:

			flag, err := f(string(delivery.Body))
			if err != nil {
				// r.failOnErr(err, "Failed to Consume")
				logx.Info(err, "Failed to Consume")
				flag = false
			}
			if flag {
				// delivery.Reject(false)
				delivery.Ack(false)
				logx.Info("confirm: Direct消息消费成功", string(delivery.Body))
			} else {
				delivery.Ack(false)
				logx.Info("confirm: Direct消息消费失败", string(delivery.Body), err)
			}
		}
		if closeFlag {
			break
		}
	}
	// 5.启用协程处理消息
	// forever := make(chan bool) // 开个channel阻塞住，让开启的协程能一直跑着
	// go func() {
	// 	for delivery := range msgs {
	// 		// 消息逻辑处理，可以自行设计逻辑
	// 		// fmt.Println("Received a message:", string(delivery.Body))
	// 		flag, err := f(string(delivery.Body))
	// 		if err != nil {
	// 			// r.failOnErr(err, "Failed to Consume")
	// 			logx.Info(err, "Failed to Consume")
	// 			flag = false
	// 		}
	// 		if flag {
	// 			// delivery.Reject(false)
	// 			delivery.Ack(false)
	// 			logx.Info("confirm: 消息消费成功", string(delivery.Body))
	// 		} else {
	// 			delivery.Ack(false)
	// 			logx.Info("confirm: 消息消费失败", string(delivery.Body), err)
	// 		}
	// 		// delivery.Ack(false)
	// 	}
	// }()
	// fmt.Println(" [*] Waiting for messages.")
	// <-forever
}

// ConsumeDead Topics模式 消费者
func (r *Rabbit) ConsumeDead(toExchange, toRouteKey string, f func(body string) (bool, error)) {
	defer func(toExchange, toRouteKey string, f func(body string) (bool, error)) {
		if err := recover(); err != nil {
			time.Sleep(3 * time.Second)
			fmt.Println("sleep: 3 seconds", err)
		}
		r = NewRabbitMQTopics(r.QueueName, r.Exchange, r.Key, r.MqUrl)
		r.ConsumeDead(toExchange, toRouteKey, f)
	}(toExchange, toRouteKey, f)
	// 1.试探性创建延迟交换机
	err := r.channel.ExchangeDeclare(
		r.Exchange,          // 交换机名字
		"x-delayed-message", // 交换机类型，这里使用topic类型，即: Topics模式
		true,                // 是否持久化
		false,               // 是否自动删除
		false,               // true表示这个exchange不可以被client用来推送消息，仅用来进行exchange和exchange之间的绑定
		false,               // 是否阻塞处理
		amqp.Table{
			"x-delayed-type": amqp.ExchangeDirect,
		}, // 额外的属性
	)
	logx.Info(err, "Failed to declare an exchange", r.Exchange)
	err = r.channel.ExchangeDeclare(
		toExchange,          // 交换机名字
		amqp.ExchangeDirect, // 交换机类型，这里使用topic类型，即: Topics模式
		true,                // 是否持久化
		false,               // 是否自动删除
		false,               // true表示这个exchange不可以被client用来推送消息，仅用来进行exchange和exchange之间的绑定
		false,               // 是否阻塞处理
		nil,                 // 额外的属性
	)
	// r.failOnErr(err, "Failed to declare an exchange")
	logx.Info(err, "Failed to declare an exchange", toExchange)
	// 2.试探性创建队列，这里注意队列名称不要写
	var args = make(map[string]interface{})
	args["x-dead-letter-exchange"] = toExchange    // 指定死信交换机
	args["x-dead-letter-routing-key"] = toRouteKey // 指定死信routing-key
	queue, err := r.channel.QueueDeclare(
		r.QueueName, // 随机生产队列名称
		true,        // 是否持久化
		false,       // 是否自动删除
		false,       // 是否具有排他性
		false,       // 是否阻塞处理
		args,        // 额外的属性
	)
	logx.Info(err, "Failed to declare a queue", r.QueueName)
	// 3.绑定队列到exchange中
	err = r.channel.QueueBind(
		queue.Name, // 队列名
		r.Key,      // 路由参数，如果匹配消息发送的时候指定的路由参数，消息就投递到当前队列（在Topics模式下，这里的key要指定）
		r.Exchange, // 交换机名字，需要跟消息发送端定义的交换器保持一致
		false,      // 是否阻塞处理
		nil,        // 额外的属性
	)
	_, err = r.channel.QueueDeclare(
		toExchange, // 随机生产队列名称
		true,       // 是否持久化
		false,      // 是否自动删除
		false,      // 是否具有排他性
		false,      // 是否阻塞处理
		nil,        // 额外的属性
	)
	err = r.channel.QueueBind(
		toExchange, // 队列名
		toRouteKey, // 路由参数，如果匹配消息发送的时候指定的路由参数，消息就投递到当前队列（在Topics模式下，这里的key要指定）
		toExchange, // 交换机名字，需要跟消息发送端定义的交换器保持一致
		false,      // 是否阻塞处理
		nil,        // 额外的属性
	)
	logx.Info(err, "Failed to declare a queue", toExchange)
	// r.failOnErr(err, "Failed to declare a queue")

	defer r.channel.Close()
	closeChan := make(chan *amqp.Error, 1)
	notifyClose := r.channel.NotifyClose(closeChan)
	closeFlag := false
	// 4.消费消息
	msgs, err := r.channel.Consume(
		queue.Name, // 队列名称
		"",         // 用来区分多个消费者
		false,      // 是否自动应答
		false,      // 是否独有
		false,      // 设置为true，表示不能将同一个Connection中生产者发送的消息传递给这个Connection中的消费者
		false,      // 队列是否阻塞
		nil,        // 额外的属性
	)
	logx.Info(err, "Failed to Consume")
	// r.failOnErr(err, "Failed to Consume")
	// 5.启用协程处理消息
	for {
		select {
		case e := <-notifyClose:
			fmt.Println("chan error,e:%s", e.Error())
			close(closeChan)
			time.Sleep(5 * time.Second)
			r = NewRabbitMQTopics(r.QueueName, r.Exchange, r.Key, r.MqUrl)
			r.ConsumeDead(toExchange, toRouteKey, f)
			closeFlag = true
		case delivery := <-msgs:
			flag, err := f(string(delivery.Body))
			if err != nil {
				// r.failOnErr(err, "Failed to Consume")
				logx.Info(err, "Failed to Consume")
				flag = false
			}
			if flag {
				// delivery.Reject(false)
				delivery.Ack(false)
				logx.Info("confirm: Dead消息消费成功", string(delivery.Body))
			} else {
				delivery.Ack(false)
				logx.Info("confirm: Dead消息消费失败", string(delivery.Body), err)
			}
		}
		if closeFlag {
			break
		}
	}
	// // 5.启用协程处理消息
	// forever := make(chan bool) // 开个channel阻塞住，让开启的协程能一直跑着
	// go func() {
	// 	for delivery := range msgs {
	// 		// 消息逻辑处理，可以自行设计逻辑
	// 		// fmt.Println("Received a message:", string(delivery.Body))
	// 		flag, err := f(string(delivery.Body))
	// 		if err != nil {
	// 			// r.failOnErr(err, "Failed to Consume")
	// 			logx.Info(err, "Failed to Consume")
	// 			flag = false
	// 		}
	// 		if flag {
	// 			// delivery.Reject(false)
	// 			delivery.Ack(false)
	// 			logx.Info("confirm: 消息消费成功", string(delivery.Body))
	// 		} else {
	// 			delivery.Reject(false)
	// 			logx.Info("confirm: 消息消费失败", string(delivery.Body), err)
	// 		}
	// 		// delivery.Ack(false)
	// 	}
	// }()
	// fmt.Println(" [*] Waiting for messages.")
	// <-forever
}
